package com.qit.prototype.flink.performance;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.TimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import com.qit.prototype.flink.entity.MarketData;
import com.qit.prototype.flink.entity.Tick;

/**
 * Simple case for data convert in Flink.
 * 
 * @author rollgan
 *
 */
public class TestCase8 {
	
	public static BlockingQueue<MarketData> outputMarketData = new LinkedBlockingQueue<MarketData>();
	
	public TestCase8() throws Exception {
		
		// Create a thread to write market data in output.
		Thread t = new Thread(new Runnable() {

			@Override
			public void run() {
				while (true) {
					try {
						MarketData md = outputMarketData.take();
						System.out.println("output: md.fieldValues=" + md.getFieldValues());
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
			}
		});
		
		t.start();

		// Create env.
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		// Env default time characteristic is processing time.
		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
		
		// Source.
		// Local entities.
		System.out.println("create source.");
		List<Tick> tickSource = new ArrayList<Tick>();
		
		for (int i = 0; i < 10000; i++) {
			Tick tick = new Tick();
			tick.setId(i + 1L);
			tick.setPrice(1.0);
			tick.setQty(i + 1.0);
			tickSource.add(tick);
		}
		
		DataStream<Tick> dsTick = env.fromCollection(tickSource);
		DataStream<MarketData> dsMarketData = dsTick.map(new MapFunction<Tick, MarketData>() {
			private static final long serialVersionUID = -6284057346699940235L;

			@Override
			public MarketData map(Tick tick) throws Exception {
				System.out.println("map");
				MarketData md = new MarketData(tick.getSymbol(), MarketData.FIELD_GROUP_TRADE);
				md.setId(tick.getId());
				md.getFieldValues().put("price", String.valueOf(tick.getPrice()));
				md.getFieldValues().put("qty", String.valueOf(tick.getQty()));
				return md;
			}
			
		});
		/*
		.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<MarketData>() {
			
			private long currentMaxTimestamp = 0l;

			@Override
			public long extractTimestamp(MarketData md,
					long previousElementTimestamp) {
				long ts = md.getId();
				currentMaxTimestamp = Math.max(currentMaxTimestamp, ts);
				return ts;
			}

			@Override
			public Watermark getCurrentWatermark() {
				return new Watermark(currentMaxTimestamp + 10);
			}
			
		});
		*/
		
		dsMarketData = dsMarketData.assignTimestampsAndWatermarks(
				WatermarkStrategy.<MarketData>forMonotonousTimestamps()
//				WatermarkStrategy.<MarketData>forBoundedOutOfOrderness(Duration.ofMillis(10))
				.withTimestampAssigner(new SerializableTimestampAssigner<MarketData>() {
					@Override
					public long extractTimestamp(MarketData element, long recordTimestamp) {
						return element.getId();
					}
				}));
		
//		WatermarkStrategy<MarketData> strategy = WatermarkStrategy
//		        .<MarketData>forBoundedOutOfOrderness(Duration.ofSeconds(1))
//		        .withTimestampAssigner((event, timestamp) -> event.getId());
		
//		dsMarketData = dsMarketData
//		.windowAll(GlobalWindows.create())
//		.process(new ProcessAllWindowFunction<MarketData, MarketData, GlobalWindow>() {
//
//			@Override
//			public void process(
//					ProcessAllWindowFunction<MarketData, MarketData, GlobalWindow>.Context arg0,
//					Iterable<MarketData> arg1, Collector<MarketData> arg2)
//					throws Exception {
//				Iterator<MarketData> it = arg1.iterator();
//				while (it.hasNext()) {
//					MarketData md = it.next();
//					outputMarketData.put(md);
//				}
//			}
//		});
		
		/*
		dsMarketData.process(new ProcessFunction<MarketData, MarketData>() {

			private static final long serialVersionUID = -2436816371621523006L;

			@Override
			public void processElement(MarketData md,
					ProcessFunction<MarketData, MarketData>.Context arg1,
					Collector<MarketData> arg2) throws Exception {
				outputMarketData.put(md);
			}
		});
		*/
		
//		dsMarketData.addSink(new SinkFunction<MarketData>() {
//			@Override
//			public void invoke(
//					MarketData md,
//					org.apache.flink.streaming.api.functions.sink.SinkFunction.Context context)
//					throws Exception {
//				System.out.println("invoke");
//				outputMarketData.put(md);
//			}
//		});
		
		Pattern<MarketData, ?> pattern = Pattern.<MarketData>begin("start")
				.where(new SimpleCondition<MarketData>() {

					private static final long serialVersionUID = -7850969134419767514L;
		
					@Override
					public boolean filter(MarketData md) throws Exception {
						return true;
					}
			
				});
		
		
		PatternStream<MarketData> patternStream = CEP.pattern(dsMarketData, pattern);
		
		patternStream.process(new PatternProcessFunction<MarketData, MarketData>() {

			private static final long serialVersionUID = 1503737342639825382L;

			@Override
			public void processMatch(
					Map<String, List<MarketData>> arg0,
					org.apache.flink.cep.functions.PatternProcessFunction.Context arg1,
					Collector<MarketData> arg2) throws Exception {
				outputMarketData.put(arg0.get("start").get(0));
			}
		});
		
		
        env.execute();
		
	}

}
